package com.red.wood.task;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import com.red.wood.model.SqlInfo;
import com.red.wood.util.CommonUtils;
import com.red.wood.util.SqlParser;
import com.red.wood.util.SysProps;

public class IpVerifyBaseTask {

	private static final Log log = LogFactory.getLog(IpVerifyBaseTask.class);
	
	public static void ipTask(SparkSession spark,SqlInfo sqlInfo,String day) {
		try {
			SysProps sysProps = SysProps.getInstance();
			
			String hive_databse = sysProps.getValue("hive.database");
			String ip_base_verify_hdfs = sysProps.getValue("ip_base_verify_hdfs");
			
			//IDC机房基础上报IP段IP数据错误核验表
			idcHouseIpErrorVerify(spark,sqlInfo,day,ip_base_verify_hdfs,hive_databse);
			
			//IDC机房基础上报IP企业间冲突核验结果表
			idcHouseIpCompanyConflictVerify(spark,sqlInfo,day,ip_base_verify_hdfs,hive_databse);
			
			//IDC活跃IP数据错误核验表
			idcActiveIpErrorVerifyTask(spark,sqlInfo,day,ip_base_verify_hdfs,hive_databse);
			
			//IDC活跃IP企业间冲突核验结果表
			idcActiveIpConflictVerify(spark,sqlInfo,day,ip_base_verify_hdfs,hive_databse);
			
			//spark.sql("insert overwrite table "+hive_databse+".t_dm_ip_verify_base partition(ds='"+day+"') select ipv4ip_begin,ipv4ip_end,ipv6ip_begin,ipv6ip_end,lyid,bbdw_id,fpfs,fpdx,fprq,wgip,wgdz,sfgn,syfs,ymtzfwq,sheng_info,ipsss,ipssx,jtiplx,ipssjf,is_cdn,is_hole,dwmc,dwxz,id_type,id_number,dwxzjb,dwfl,dwhyfl,shengid,shiid,xianid,xxdz,lxrxm,lxrdh,lxrdzyj,jyxkzh,protocol,idc_last_active_time,idc_active_visit_count,dns_visit_count,source_id,isconflict,conflict_content,insert_time,operators,bz,application_service_type,ipzqdbl,ipkxdbl,country,iplatitude,iplongitude,ircs_id,cdn_id,dns_id,housetype,is_idcisp,is_zx,is_ircs,is_dns_server,cdn_active_visit_count,ircs_live_visit_count,ip_service_type,bbdw_provinceid,bbdw_cityid,bbdw_countyid,bbdw_name,distribute_name,source_name,house_name,house_province,house_city,house_county,is_idc_active,is_ircs_active,is_cdn_active,is_dns_active,ircs_name,cdn_name,ircs_provinceid,verify_type,yl3 from "+hive_databse+".t_dm_ip_verify_base where 1 = 2");
			long count = spark.sql("select ds from "+hive_databse+".t_dm_ip_verify_base where ds='"+day+"' limit 1").count();
			if(count > 0) {
				spark.sql("truncate table "+hive_databse+".t_dm_ip_verify_base partition(ds='"+day+"')");
			}
			String dwdPath= ip_base_verify_hdfs+"/*";
			String sql ="load data inpath '"+dwdPath+"/*' into table "+hive_databse+".t_dm_ip_verify_base partition (ds='"+day+"')";
			
			spark.sql(sql);
		}catch(Exception e) {
			if(log.isErrorEnabled()) {
				log.error(e.getMessage(),e);
			}
		}
	}
	
	private static void idcHouseIpCompanyConflictVerify(SparkSession spark,SqlInfo sqlInfo,String day,String ip_base_verify_hdfs,String hive_databse) {
		
		String sqlIdcHouseIpCompanyConflictVerify = sqlInfo.getSqlIdcHouseIpCompanyConflictVerify();
		sqlIdcHouseIpCompanyConflictVerify = SqlParser.getOriginalOutputSql(sqlIdcHouseIpCompanyConflictVerify,"hive_databse",hive_databse);
		sqlIdcHouseIpCompanyConflictVerify = SqlParser.getSql(sqlIdcHouseIpCompanyConflictVerify, "query_day", day);
		
		Dataset<Row> dataset = spark.sql(sqlIdcHouseIpCompanyConflictVerify);
		
		JavaRDD<String> rdd = dataset.toJavaRDD().map(new Function<Row, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public String call(Row row){
				StringBuffer buffer = new StringBuffer();
				try {
					String ipv4ip_begin = CommonUtils.valueOf(row.getAs("ipv4ip_begin"));
					String ipv4ip_end = CommonUtils.valueOf(row.getAs("ipv4ip_end"));
					String ipv6ip_begin = "";
					String ipv6ip_end = "";
					String lyid = "";
					String bbdw_id = CommonUtils.valueOf(row.getAs("bbdw_id"));
					String fpfs = "";
					String fpdx = "";
					String fprq = CommonUtils.valueOf(row.getAs("fprq"));
					String wgip = "";
					String wgdz = "";
					String sfgn = "";
					String syfs = CommonUtils.valueOf(row.getAs("syfs"));;
					String ymtzfwq = "";
					String sheng_info = "";
					String ipsss = "";
					String ipssx = "";
					String jtiplx = "";
					String ipssjf = CommonUtils.valueOf(row.getAs("ipssjf"));
					String is_cdn = "";
					String is_hole = "";
					String dwmc = CommonUtils.valueOf(row.getAs("dwmc"));
					String dwxz = "";
					String id_type = CommonUtils.valueOf(row.getAs("id_type"));
					String id_number = CommonUtils.valueOf(row.getAs("id_number"));
					String dwxzjb = "";
					String dwfl = "";
					String dwhyfl = "";
					String shengid = "";
					String shiid = "";
					String xianid = "";
					String xxdz = "";
					String lxrxm = "";
					String lxrdh = "";
					String lxrdzyj = "";
					String jyxkzh = "";
					String protocol = "";
					String idc_last_active_time = "";
					String idc_active_visit_count = "";
					String dns_visit_count = "";
					String source_id = "";
					String isconflict = CommonUtils.valueOf(row.getAs("isconflict"));
					String conflict_content = "";
					String insert_time = "";
					String operators = "";
					String bz = "";
					String application_service_type = "";
					String ipzqdbl = "";
					String ipkxdbl = "";
					String country = "";
					String iplatitude = "";
					String iplongitude = "";
					String ircs_id = "";
					String cdn_id = "";
					String dns_id = "";
					String housetype = "";
					String is_idcisp = CommonUtils.valueOf(row.getAs("is_idcisp"));
					String is_zx = CommonUtils.valueOf(row.getAs("is_zx"));
					String is_ircs = "";
					String is_dns_server = "";
					String cdn_active_visit_count = "";
					String ircs_live_visit_count = "";
					String ip_service_type = "";
					String bbdw_provinceId = CommonUtils.valueOf(row.getAs("bbdw_provinceId"));
					String bbdw_cityId = "";
					String bbdw_countyId = "";
					String bbdw_name = "";
					String distribute_name = "";
					String source_name = "";
					String house_name = "";
					String house_province = "";
					String house_city = "";	
					String house_county = "";
					String is_idc_active = "";
					String is_ircs_active = "";
					String is_cdn_active = "";
					String is_dns_active = "";
					String ircs_name = "";
					String cdn_name = "";
					String ircs_provinceId = "";
					String verify_type = CommonUtils.valueOf(row.getAs("verify_type"));
					String yl3 = "";
					
					buffer.append(ipv4ip_begin+"\t").append(ipv4ip_end+"\t").append(ipv6ip_begin+"\t").append(ipv6ip_end+"\t").append(lyid+"\t")
						  .append(bbdw_id+"\t").append(fpfs+"\t").append(fpdx+"\t").append(fprq+"\t").append(wgip+"\t")
						  .append(wgdz+"\t").append(sfgn+"\t").append(syfs+"\t").append(ymtzfwq+"\t").append(sheng_info+"\t")
						  .append(ipsss+"\t").append(ipssx+"\t").append(jtiplx+"\t").append(ipssjf+"\t").append(is_cdn+"\t")
						  .append(is_hole+"\t").append(dwmc+"\t").append(dwxz+"\t").append(id_type+"\t").append(id_number+"\t")
						  .append(dwxzjb+"\t").append(dwfl+"\t").append(dwhyfl+"\t").append(shengid+"\t").append(shiid+"\t")
						  .append(xianid+"\t").append(xxdz+"\t").append(lxrxm+"\t").append(lxrdh+"\t").append(lxrdzyj+"\t")
						  .append(jyxkzh+"\t").append(protocol+"\t").append(idc_last_active_time+"\t").append(idc_active_visit_count+"\t").append(dns_visit_count+"\t")
						  .append(source_id+"\t").append(isconflict+"\t").append(conflict_content+"\t").append(insert_time+"\t").append(operators+"\t")
						  .append(bz+"\t").append(application_service_type+"\t").append(ipzqdbl+"\t").append(ipkxdbl+"\t").append(country+"\t")
						  .append(iplatitude+"\t").append(iplongitude+"\t").append(ircs_id+"\t").append(cdn_id+"\t").append(dns_id+"\t")
						  .append(housetype+"\t").append(is_idcisp+"\t").append(is_zx+"\t").append(is_ircs+"\t").append(is_dns_server+"\t")
						  .append(cdn_active_visit_count+"\t").append(ircs_live_visit_count+"\t").append(ip_service_type+"\t").append(bbdw_provinceId+"\t").append(bbdw_cityId+"\t")
						  .append(bbdw_countyId+"\t").append(bbdw_name+"\t").append(distribute_name+"\t").append(source_name+"\t").append(house_name+"\t")
						  .append(house_province+"\t").append(house_city+"\t").append(house_county+"\t").append(is_idc_active+"\t").append(is_ircs_active+"\t")
						  .append(is_cdn_active+"\t").append(is_dns_active+"\t").append(ircs_name+"\t").append(cdn_name+"\t").append(ircs_provinceId+"\t")
						  .append(verify_type+"\t").append(yl3+"\t");
				}catch(Exception e) {
					if(log.isErrorEnabled()) {
						log.error(e.getMessage(),e);
					}
				}
				return buffer.toString();
			}
		});
		
		if(rdd.count() > 0) {
			String dwdPath= ip_base_verify_hdfs+"/idcHouseIpCompanyConflictVerify";
			CommonUtils.uploadHdfs(rdd,dwdPath);
		}
	}
	
	private static void idcHouseIpErrorVerify(SparkSession spark,SqlInfo sqlInfo,String day,String ip_base_verify_hdfs,String hive_databse) {
		
		String sqlIdcHouseIpErrorVerify = sqlInfo.getSqlIdcHouseIpErrorVerify();
		sqlIdcHouseIpErrorVerify = SqlParser.getOriginalOutputSql(sqlIdcHouseIpErrorVerify,"hive_databse",hive_databse);
		sqlIdcHouseIpErrorVerify = SqlParser.getSql(sqlIdcHouseIpErrorVerify, "query_day", day);
		
		Dataset<Row> dataset = spark.sql(sqlIdcHouseIpErrorVerify);
		
		JavaRDD<String> rdd = dataset.toJavaRDD().map(new Function<Row, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public String call(Row row){
				StringBuffer buffer = new StringBuffer();
				try {
					String ipv4ip_begin = CommonUtils.valueOf(row.getAs("ipv4ip_begin"));
					String ipv4ip_end = CommonUtils.valueOf(row.getAs("ipv4ip_end"));
					String ipv6ip_begin = "";
					String ipv6ip_end = "";
					String lyid = "";
					String bbdw_id = CommonUtils.valueOf(row.getAs("bbdw_id"));
					String fpfs = "";
					String fpdx = "";
					String fprq = CommonUtils.valueOf(row.getAs("fprq"));
					String wgip = "";
					String wgdz = "";
					String sfgn = "";
					String syfs = CommonUtils.valueOf(row.getAs("syfs"));;
					String ymtzfwq = "";
					String sheng_info = "";
					String ipsss = "";
					String ipssx = "";
					String jtiplx = "";
					String ipssjf = CommonUtils.valueOf(row.getAs("ipssjf"));
					String is_cdn = "";
					String is_hole = "";
					String dwmc = CommonUtils.valueOf(row.getAs("dwmc"));
					String dwxz = "";
					String id_type = CommonUtils.valueOf(row.getAs("id_type"));
					String id_number = CommonUtils.valueOf(row.getAs("id_number"));
					String dwxzjb = "";
					String dwfl = "";
					String dwhyfl = "";
					String shengid = "";
					String shiid = "";
					String xianid = "";
					String xxdz = "";
					String lxrxm = "";
					String lxrdh = "";
					String lxrdzyj = "";
					String jyxkzh = "";
					String protocol = "";
					String idc_last_active_time = "";
					String idc_active_visit_count = "";
					String dns_visit_count = "";
					String source_id = "";
					String isconflict = "";
					String conflict_content = "";
					String insert_time = "";
					String operators = "";
					String bz = "";
					String application_service_type = "";
					String ipzqdbl = "";
					String ipkxdbl = "";
					String country = "";
					String iplatitude = "";
					String iplongitude = "";
					String ircs_id = "";
					String cdn_id = "";
					String dns_id = "";
					String housetype = "";
					String is_idcisp = CommonUtils.valueOf(row.getAs("is_idcisp"));
					String is_zx = CommonUtils.valueOf(row.getAs("is_zx"));
					String is_ircs = "";
					String is_dns_server = "";
					String cdn_active_visit_count = "";
					String ircs_live_visit_count = "";
					String ip_service_type = "";
					String bbdw_provinceId = CommonUtils.valueOf(row.getAs("bbdw_provinceId"));
					String bbdw_cityId = "";
					String bbdw_countyId = "";
					String bbdw_name = "";
					String distribute_name = "";
					String source_name = "";
					String house_name = "";
					String house_province = "";
					String house_city = "";	
					String house_county = "";
					String is_idc_active = "";
					String is_ircs_active = "";
					String is_cdn_active = "";
					String is_dns_active = "";
					String ircs_name = "";
					String cdn_name = "";
					String ircs_provinceId = "";
					String verify_type = CommonUtils.valueOf(row.getAs("verify_type"));
					String yl3 = "";
					
					buffer.append(ipv4ip_begin+"\t").append(ipv4ip_end+"\t").append(ipv6ip_begin+"\t").append(ipv6ip_end+"\t").append(lyid+"\t")
						  .append(bbdw_id+"\t").append(fpfs+"\t").append(fpdx+"\t").append(fprq+"\t").append(wgip+"\t")
						  .append(wgdz+"\t").append(sfgn+"\t").append(syfs+"\t").append(ymtzfwq+"\t").append(sheng_info+"\t")
						  .append(ipsss+"\t").append(ipssx+"\t").append(jtiplx+"\t").append(ipssjf+"\t").append(is_cdn+"\t")
						  .append(is_hole+"\t").append(dwmc+"\t").append(dwxz+"\t").append(id_type+"\t").append(id_number+"\t")
						  .append(dwxzjb+"\t").append(dwfl+"\t").append(dwhyfl+"\t").append(shengid+"\t").append(shiid+"\t")
						  .append(xianid+"\t").append(xxdz+"\t").append(lxrxm+"\t").append(lxrdh+"\t").append(lxrdzyj+"\t")
						  .append(jyxkzh+"\t").append(protocol+"\t").append(idc_last_active_time+"\t").append(idc_active_visit_count+"\t").append(dns_visit_count+"\t")
						  .append(source_id+"\t").append(isconflict+"\t").append(conflict_content+"\t").append(insert_time+"\t").append(operators+"\t")
						  .append(bz+"\t").append(application_service_type+"\t").append(ipzqdbl+"\t").append(ipkxdbl+"\t").append(country+"\t")
						  .append(iplatitude+"\t").append(iplongitude+"\t").append(ircs_id+"\t").append(cdn_id+"\t").append(dns_id+"\t")
						  .append(housetype+"\t").append(is_idcisp+"\t").append(is_zx+"\t").append(is_ircs+"\t").append(is_dns_server+"\t")
						  .append(cdn_active_visit_count+"\t").append(ircs_live_visit_count+"\t").append(ip_service_type+"\t").append(bbdw_provinceId+"\t").append(bbdw_cityId+"\t")
						  .append(bbdw_countyId+"\t").append(bbdw_name+"\t").append(distribute_name+"\t").append(source_name+"\t").append(house_name+"\t")
						  .append(house_province+"\t").append(house_city+"\t").append(house_county+"\t").append(is_idc_active+"\t").append(is_ircs_active+"\t")
						  .append(is_cdn_active+"\t").append(is_dns_active+"\t").append(ircs_name+"\t").append(cdn_name+"\t").append(ircs_provinceId+"\t")
						  .append(verify_type+"\t").append(yl3+"\t");
				}catch(Exception e) {
					if(log.isErrorEnabled()) {
						log.error(e.getMessage(),e);
					}
				}
				return buffer.toString();
			}
		});
		
		if(rdd.count() > 0) {
			String dwdPath= ip_base_verify_hdfs+"/idcHouseIpErrorVerify";
			CommonUtils.uploadHdfs(rdd,dwdPath);
		}
		
	}
	
	private static void idcActiveIpConflictVerify(SparkSession spark,SqlInfo sqlInfo,String day,String ip_base_verify_hdfs,String hive_databse) {
		
		String sqlIdcActiveIpConflictVerify = sqlInfo.getSqlIdcActiveIpConflictVerify();
		sqlIdcActiveIpConflictVerify = SqlParser.getOriginalOutputSql(sqlIdcActiveIpConflictVerify,"hive_databse",hive_databse);
		sqlIdcActiveIpConflictVerify = SqlParser.getSql(sqlIdcActiveIpConflictVerify, "query_day", day);
		
		Dataset<Row> dataset = spark.sql(sqlIdcActiveIpConflictVerify);
		
		JavaRDD<String> rdd = dataset.toJavaRDD().map(new Function<Row, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public String call(Row row){
				StringBuffer buffer = new StringBuffer();
				try {
					String ipv4ip_begin = CommonUtils.valueOf(row.getAs("ipv4ip_begin"));
					String ipv4ip_end = CommonUtils.valueOf(row.getAs("ipv4ip_end"));
					String ipv6ip_begin = "";
					String ipv6ip_end = "";
					String lyid = "";
					String bbdw_id = CommonUtils.valueOf(row.getAs("bbdw_id"));
					String fpfs = "";
					String fpdx = "";
					String fprq = "";
					String wgip = "";
					String wgdz = "";
					String sfgn = "";
					String syfs = "";
					String ymtzfwq = "";
					String sheng_info = "";
					String ipsss = "";
					String ipssx = "";
					String jtiplx = "";
					String ipssjf = CommonUtils.valueOf(row.getAs("ipssjf"));
					String is_cdn = "";
					String is_hole = "";
					String dwmc = "";
					String dwxz = "";
					String id_type = "";
					String id_number = "";
					String dwxzjb = "";
					String dwfl = "";
					String dwhyfl = "";
					String shengid = "";
					String shiid = "";
					String xianid = "";
					String xxdz = "";
					String lxrxm = "";
					String lxrdh = "";
					String lxrdzyj = "";
					String jyxkzh = "";
					String protocol = CommonUtils.valueOf(row.getAs("protocol"));
					String idc_last_active_time = CommonUtils.valueOf(row.getAs("idc_last_active_time"));
					String idc_active_visit_count = CommonUtils.valueOf(row.getAs("idc_active_visit_count"));
					String dns_visit_count = "";
					String source_id = "";
					String isconflict = CommonUtils.valueOf(row.getAs("isconflict"));
					String conflict_content = "";
					String insert_time = "";
					String operators = "";
					String bz = "";
					String application_service_type = "";
					String ipzqdbl = "";
					String ipkxdbl = "";
					String country = "";
					String iplatitude = "";
					String iplongitude = "";
					String ircs_id = "";
					String cdn_id = "";
					String dns_id = "";
					String housetype = "";
					String is_idcisp = CommonUtils.valueOf(row.getAs("is_idcisp"));
					String is_zx = CommonUtils.valueOf(row.getAs("is_zx"));
					String is_ircs = "";
					String is_dns_server = "";
					String cdn_active_visit_count = "";
					String ircs_live_visit_count = "";
					String ip_service_type = "";
					String bbdw_provinceId = CommonUtils.valueOf(row.getAs("bbdw_provinceId"));
					String bbdw_cityId = "";
					String bbdw_countyId = "";
					String bbdw_name = "";
					String distribute_name = "";
					String source_name = "";
					String house_name = "";
					String house_province = "";
					String house_city = "";	
					String house_county = "";
					String is_idc_active = "";
					String is_ircs_active = "";
					String is_cdn_active = "";
					String is_dns_active = "";
					String ircs_name = "";
					String cdn_name = "";
					String ircs_provinceId = "";
					String verify_type = CommonUtils.valueOf(row.getAs("verify_type"));
					String yl3 = "";
					
					buffer.append(ipv4ip_begin+"\t").append(ipv4ip_end+"\t").append(ipv6ip_begin+"\t").append(ipv6ip_end+"\t").append(lyid+"\t")
						  .append(bbdw_id+"\t").append(fpfs+"\t").append(fpdx+"\t").append(fprq+"\t").append(wgip+"\t")
						  .append(wgdz+"\t").append(sfgn+"\t").append(syfs+"\t").append(ymtzfwq+"\t").append(sheng_info+"\t")
						  .append(ipsss+"\t").append(ipssx+"\t").append(jtiplx+"\t").append(ipssjf+"\t").append(is_cdn+"\t")
						  .append(is_hole+"\t").append(dwmc+"\t").append(dwxz+"\t").append(id_type+"\t").append(id_number+"\t")
						  .append(dwxzjb+"\t").append(dwfl+"\t").append(dwhyfl+"\t").append(shengid+"\t").append(shiid+"\t")
						  .append(xianid+"\t").append(xxdz+"\t").append(lxrxm+"\t").append(lxrdh+"\t").append(lxrdzyj+"\t")
						  .append(jyxkzh+"\t").append(protocol+"\t").append(idc_last_active_time+"\t").append(idc_active_visit_count+"\t").append(dns_visit_count+"\t")
						  .append(source_id+"\t").append(isconflict+"\t").append(conflict_content+"\t").append(insert_time+"\t").append(operators+"\t")
						  .append(bz+"\t").append(application_service_type+"\t").append(ipzqdbl+"\t").append(ipkxdbl+"\t").append(country+"\t")
						  .append(iplatitude+"\t").append(iplongitude+"\t").append(ircs_id+"\t").append(cdn_id+"\t").append(dns_id+"\t")
						  .append(housetype+"\t").append(is_idcisp+"\t").append(is_zx+"\t").append(is_ircs+"\t").append(is_dns_server+"\t")
						  .append(cdn_active_visit_count+"\t").append(ircs_live_visit_count+"\t").append(ip_service_type+"\t").append(bbdw_provinceId+"\t").append(bbdw_cityId+"\t")
						  .append(bbdw_countyId+"\t").append(bbdw_name+"\t").append(distribute_name+"\t").append(source_name+"\t").append(house_name+"\t")
						  .append(house_province+"\t").append(house_city+"\t").append(house_county+"\t").append(is_idc_active+"\t").append(is_ircs_active+"\t")
						  .append(is_cdn_active+"\t").append(is_dns_active+"\t").append(ircs_name+"\t").append(cdn_name+"\t").append(ircs_provinceId+"\t")
						  .append(verify_type+"\t").append(yl3+"\t");
				}catch(Exception e) {
					if(log.isErrorEnabled()) {
						log.error(e.getMessage(),e);
					}
				}
				return buffer.toString();
			}
		});
		
		if(rdd.count() > 0) {
			String dwdPath= ip_base_verify_hdfs+"/idcActiveIpConflictVerify";
			CommonUtils.uploadHdfs(rdd,dwdPath);
		}
	}

	private static void idcActiveIpErrorVerifyTask(SparkSession spark,SqlInfo sqlInfo,String day,String ip_base_verify_hdfs,String hive_databse) {
		
		String sqlIdcActiveIpErrorVerify = sqlInfo.getSqlIdcActiveIpErrorVerify();
		sqlIdcActiveIpErrorVerify = SqlParser.getOriginalOutputSql(sqlIdcActiveIpErrorVerify,"hive_databse",hive_databse);
		sqlIdcActiveIpErrorVerify = SqlParser.getSql(sqlIdcActiveIpErrorVerify, "query_day", day);
		
		Dataset<Row> dataset = spark.sql(sqlIdcActiveIpErrorVerify);
		
		JavaRDD<String> rdd = dataset.toJavaRDD().map(new Function<Row, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public String call(Row row){
				StringBuffer buffer = new StringBuffer();
				try {
					String ipv4ip_begin = CommonUtils.valueOf(row.getAs("ipv4ip_begin"));
					String ipv4ip_end = CommonUtils.valueOf(row.getAs("ipv4ip_end"));
					String ipv6ip_begin = "";
					String ipv6ip_end = "";
					String lyid = "";
					String bbdw_id = CommonUtils.valueOf(row.getAs("bbdw_id"));
					String fpfs = "";
					String fpdx = "";
					String fprq = "";
					String wgip = "";
					String wgdz = "";
					String sfgn = "";
					String syfs = "";
					String ymtzfwq = "";
					String sheng_info = "";
					String ipsss = "";
					String ipssx = "";
					String jtiplx = "";
					String ipssjf = CommonUtils.valueOf(row.getAs("ipssjf"));
					String is_cdn = "";
					String is_hole = "";
					String dwmc = "";
					String dwxz = "";
					String id_type = "";
					String id_number = "";
					String dwxzjb = "";
					String dwfl = "";
					String dwhyfl = "";
					String shengid = "";
					String shiid = "";
					String xianid = "";
					String xxdz = "";
					String lxrxm = "";
					String lxrdh = "";
					String lxrdzyj = "";
					String jyxkzh = "";
					String protocol = CommonUtils.valueOf(row.getAs("protocol"));
					String idc_last_active_time = CommonUtils.valueOf(row.getAs("idc_last_active_time"));
					String idc_active_visit_count = CommonUtils.valueOf(row.getAs("idc_active_visit_count"));
					String dns_visit_count = "";
					String source_id = "";
					String isconflict = "";
					String conflict_content = "";
					String insert_time = "";
					String operators = "";
					String bz = "";
					String application_service_type = "";
					String ipzqdbl = "";
					String ipkxdbl = "";
					String country = "";
					String iplatitude = "";
					String iplongitude = "";
					String ircs_id = "";
					String cdn_id = "";
					String dns_id = "";
					String housetype = "";
					String is_idcisp = CommonUtils.valueOf(row.getAs("is_idcisp"));
					String is_zx = CommonUtils.valueOf(row.getAs("is_zx"));
					String is_ircs = "";
					String is_dns_server = "";
					String cdn_active_visit_count = "";
					String ircs_live_visit_count = "";
					String ip_service_type = "";
					String bbdw_provinceId = CommonUtils.valueOf(row.getAs("bbdw_provinceId"));
					String bbdw_cityId = "";
					String bbdw_countyId = "";
					String bbdw_name = "";
					String distribute_name = "";
					String source_name = "";
					String house_name = "";
					String house_province = "";
					String house_city = "";	
					String house_county = "";
					String is_idc_active = "";
					String is_ircs_active = "";
					String is_cdn_active = "";
					String is_dns_active = "";
					String ircs_name = "";
					String cdn_name = "";
					String ircs_provinceId = "";
					String verify_type = CommonUtils.valueOf(row.getAs("verify_type"));
					String yl3 = "";
					
					buffer.append(ipv4ip_begin+"\t").append(ipv4ip_end+"\t").append(ipv6ip_begin+"\t").append(ipv6ip_end+"\t").append(lyid+"\t")
						  .append(bbdw_id+"\t").append(fpfs+"\t").append(fpdx+"\t").append(fprq+"\t").append(wgip+"\t")
						  .append(wgdz+"\t").append(sfgn+"\t").append(syfs+"\t").append(ymtzfwq+"\t").append(sheng_info+"\t")
						  .append(ipsss+"\t").append(ipssx+"\t").append(jtiplx+"\t").append(ipssjf+"\t").append(is_cdn+"\t")
						  .append(is_hole+"\t").append(dwmc+"\t").append(dwxz+"\t").append(id_type+"\t").append(id_number+"\t")
						  .append(dwxzjb+"\t").append(dwfl+"\t").append(dwhyfl+"\t").append(shengid+"\t").append(shiid+"\t")
						  .append(xianid+"\t").append(xxdz+"\t").append(lxrxm+"\t").append(lxrdh+"\t").append(lxrdzyj+"\t")
						  .append(jyxkzh+"\t").append(protocol+"\t").append(idc_last_active_time+"\t").append(idc_active_visit_count+"\t").append(dns_visit_count+"\t")
						  .append(source_id+"\t").append(isconflict+"\t").append(conflict_content+"\t").append(insert_time+"\t").append(operators+"\t")
						  .append(bz+"\t").append(application_service_type+"\t").append(ipzqdbl+"\t").append(ipkxdbl+"\t").append(country+"\t")
						  .append(iplatitude+"\t").append(iplongitude+"\t").append(ircs_id+"\t").append(cdn_id+"\t").append(dns_id+"\t")
						  .append(housetype+"\t").append(is_idcisp+"\t").append(is_zx+"\t").append(is_ircs+"\t").append(is_dns_server+"\t")
						  .append(cdn_active_visit_count+"\t").append(ircs_live_visit_count+"\t").append(ip_service_type+"\t").append(bbdw_provinceId+"\t").append(bbdw_cityId+"\t")
						  .append(bbdw_countyId+"\t").append(bbdw_name+"\t").append(distribute_name+"\t").append(source_name+"\t").append(house_name+"\t")
						  .append(house_province+"\t").append(house_city+"\t").append(house_county+"\t").append(is_idc_active+"\t").append(is_ircs_active+"\t")
						  .append(is_cdn_active+"\t").append(is_dns_active+"\t").append(ircs_name+"\t").append(cdn_name+"\t").append(ircs_provinceId+"\t")
						  .append(verify_type+"\t").append(yl3+"\t");
				}catch(Exception e) {
					if(log.isErrorEnabled()) {
						log.error(e.getMessage(),e);
					}
				}
				return buffer.toString();
			}
		});
		
		if(rdd.count() > 0) {
			String dwdPath= ip_base_verify_hdfs+"/idcActiveIpErrorVerify";
			CommonUtils.uploadHdfs(rdd,dwdPath);
		}
	}
}